package com.niodata.dt.fs.us3;

import cn.ucloud.ufile.UfileClient;
import cn.ucloud.ufile.api.object.ObjectApiBuilder;
import cn.ucloud.ufile.api.object.ObjectConfig;
import cn.ucloud.ufile.api.object.ObjectListApi;
import cn.ucloud.ufile.api.object.ObjectListWithDirFormatApi;
import cn.ucloud.ufile.auth.ObjectAuthorization;
import cn.ucloud.ufile.auth.UfileObjectLocalAuthorization;
import cn.ucloud.ufile.bean.CommonPrefix;
import cn.ucloud.ufile.bean.ObjectContentBean;
import cn.ucloud.ufile.bean.ObjectInfoBean;
import cn.ucloud.ufile.bean.ObjectListBean;
import cn.ucloud.ufile.bean.ObjectListWithDirFormatBean;
import cn.ucloud.ufile.bean.ObjectProfile;
import cn.ucloud.ufile.util.UfileProtocol;
import com.google.common.base.Strings;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.util.LinkedList;
import java.util.List;

import com.niodata.dt.fs.ArrayUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.util.Progressable;
import org.apache.log4j.Logger;

public class Us3FileSystem extends FileSystem {

  private static final Logger logger = Logger.getLogger(Us3FileSystem.class);
  private ObjectAuthorization objectAuthorization;
  private ObjectConfig config;
  private String bucket;
  private String region;
  private String endPoint;
  private URI uri;
  private Path workingDir;

  /**
   * Called after a new FileSystem instance is constructed.
   *
   * @param name a uri whose authority section names the host, port, etc. for this FileSystem
   * @param conf the configuration
   */
  public void initialize(URI name, Configuration conf) throws IOException {
    super.setConf(conf);
    super.initialize(name, conf);
    this.bucket = conf.get(ConfigKeys.BUCKET);
    this.region = conf.get(ConfigKeys.REGION);
    this.endPoint = conf.get(ConfigKeys.ENDPOINT, "ufileos.com");
    objectAuthorization = new UfileObjectLocalAuthorization(
          conf.get(ConfigKeys.ACCESS_KEY), conf.get(ConfigKeys.SECRET_KEY));
    config = new ObjectConfig(region, endPoint);
    String https = conf.get(ConfigKeys.PROTOCOL, "https");
    if ("https".equals(https)) {
      config.withProtocol(UfileProtocol.PROTOCOL_HTTPS);
    } else {
      config.withProtocol(UfileProtocol.PROTOCOL_HTTP);
    }
    String url = "us3://" + bucket + "." + region + "." + endPoint;
    uri = URI.create(url);
    workingDir = new Path(System.getProperty("user.dir"));

  }

  @Override
  public URI getUri() {
    return uri;
  }

  @Override
  public FSDataInputStream open(Path f, int bufferSize) throws IOException {
    FileStatus status = getFileStatus(f);
    if (status.isDirectory()) {
      throw new IOException("path is directory," + f);
    }
    ObjectApiBuilder apiBuilder = UfileClient.object(objectAuthorization, config);
    try {
      //apiBuilder.downloadFile(null).withinRange()
      String key = pathToKey(f);
      ObjectProfile profile = apiBuilder.objectProfile(key, bucket).execute();
      Us3InputStream us3InputStream = new Us3InputStream(profile, apiBuilder.downloadFile(profile));
      return new FSDataInputStream(us3InputStream);
    } catch (Exception e) {
      throw new IOException(e);
    }
  }

  @Override
  public FSDataOutputStream create(Path f, FsPermission permission, boolean overwrite,
        int bufferSize, short replication, long blockSize, Progressable progress)
        throws IOException {
    String key = pathToKey(f);
    try {
      FileStatus status = getFileStatus(f);
      if (status.isDirectory()) {
        throw new IOException("path is directory," + f);
      }
      if (!overwrite) {
        throw new IOException("file exists," + f);
      }
    } catch (FileNotFoundException ex) {
      //nothing to do
    }

    FileStatus ps = getFileStatus(f.getParent());
    if (ps.isFile()) {
      throw new IOException("path is file:" + ps.getPath());
    }

    ObjectApiBuilder apiBuilder = UfileClient.object(objectAuthorization, config);
    Us3Uploader uploader = new Us3Uploader(bucket, key, apiBuilder);
    FSDataOutputStream outputStream = new FSDataOutputStream(new Us3OutputStream(uploader), null);
    return outputStream;
  }

  @Override
  public void copyFromLocalFile(boolean delSrc, boolean overwrite, Path src,
        Path dst) throws IOException {
    File local = new File(Path.getPathWithoutSchemeAndAuthority(src).toString());
    if (!local.exists() || local.isDirectory()) {
      throw new IOException("src file is directory or not exists");
    }
    String key = pathToKey(dst);
    if (exists(dst)) {
      if (getFileStatus(dst).isFile()) {
        if (!overwrite) {
          throw new IOException("dst file exists");
        }
      } else {
        key = key + "/" + src.getName();
      }
    } else {
      Path parent = dst.getParent();
      FileStatus ps = getFileStatus(parent);
      if (ps.isFile()) {
        throw new IOException("dst parent is file");
      }
    }
    ObjectApiBuilder apiBuilder = UfileClient.object(objectAuthorization, config);
    Us3Uploader uploader = new Us3Uploader(bucket, key, apiBuilder);
    uploader.put(local);
    if (delSrc) {
      local.delete();
    }
  }

  @Override
  public void copyToLocalFile(boolean delSrc, Path src, Path dst) throws IOException {
    ObjectApiBuilder apiBuilder = UfileClient.object(objectAuthorization, config);
    if (!exists(src) || !isFile(src)) {
      throw new IOException("src is directory  or file not exists:" + src);
    }
    String key = pathToKey(src);
    try {
      ObjectProfile profile = apiBuilder.objectProfile(key, bucket).execute();
      File dstFile = new File(dst.toString());
      File dstParentFile = new File(dst.getParent().toString());
      if (dstFile.exists()) {
        //dst is file : overwrite
        //dst is dir: create file under it
        if (dstFile.isDirectory()) {
          apiBuilder.downloadFile(profile).saveAt(dstFile.getAbsolutePath(),
                src.getName()).execute();
        } else {
          apiBuilder.downloadFile(profile).saveAt(dstParentFile.getAbsolutePath(),
                dst.getName()).execute();
        }
      } else {
        //if (dstParentFile.getAbsolutePath().exists() && dstParentFile.isDirectory()) {
        File dstParentFilePath = new File(dstParentFile.getAbsolutePath());
        if (dstParentFilePath.exists()) {
          apiBuilder.downloadFile(profile).saveAt(dstParentFile.getAbsolutePath(),
                dst.getName()).execute();
        } else {
          throw new FileNotFoundException(dstParentFile.getAbsolutePath());
        }
      }
    } catch (Exception ex) {
      throw new IOException(ex);
    }

  }

  @Override
  public FSDataOutputStream append(Path f, int bufferSize, Progressable progress)
        throws IOException {

    throw new IOException("Append operation Not supported");

  }

  /**
   * rename path.
   *
   * @param src src path
   * @param dst dst path
   * @return success  or failed
   * @throws IOException e
   */
  @Override
  public boolean rename(Path src, Path dst) throws IOException {
    ObjectApiBuilder apiBuilder = UfileClient.object(objectAuthorization, config);
    if (exists(dst)) {
      throw new IOException("dst file exists:" + dst);
    }
    FileStatus srcStatus = getFileStatus(src);
    String srcKey = Path.getPathWithoutSchemeAndAuthority(srcStatus.getPath()).toString();
    if (srcKey.startsWith("/")) {
      srcKey = srcKey.substring(1);
    }
    if (srcStatus.isFile()) {
      if (srcKey.endsWith("/")) {
        srcKey = srcKey.substring(0, srcKey.length() - 1);
      }
    } else {
      if (!srcKey.endsWith("/")) {
        srcKey = srcKey + "/";
      }
    }
    String dstKey = Path.getPathWithoutSchemeAndAuthority(dst).toString();
    if (dstKey.startsWith("/")) {
      dstKey = dstKey.substring(1);
    }
    if (srcStatus.isFile()) {
      if (dstKey.endsWith("/")) {
        dstKey = dstKey.substring(0, dstKey.length() - 1);
      }
    } else {
      if (!dstKey.endsWith("/")) {
        dstKey = dstKey + "/";
      }
    }
    try {
      if (srcStatus.isFile()) {
        apiBuilder.renameObject(bucket, srcKey).isRenamedTo(dstKey).execute();
      } else {
        FileStatus[] statuses = listStatus(src);
        // if (statuses.length == 1) {
        apiBuilder.renameObject(bucket, srcKey).isRenamedTo(dstKey).execute();
        //} else {
        //  throw new IOException("src path not empty");
        //}
      }
    } catch (Exception e) {
      throw new IOException(e);
    }
    return true;
  }

  /**
   * delete path.
   *
   * @param f file path
   * @param recursive recursive
   * @return success
   * @throws IOException ex
   */
  @Override
  public boolean delete(Path f, boolean recursive) throws IOException {
    if (!exists(f)) {
      return true;
    }
    FileStatus status = getFileStatus(f);
    if (status.isFile()) {
      ObjectApiBuilder apiBuilder = UfileClient.object(objectAuthorization, config);
      String key = Path.getPathWithoutSchemeAndAuthority(f).toString();
      if (key.startsWith("/")) {
        key = key.substring(1);
      }
      logger.info("delete file :" + key);
      deleteKey(key);
    } else {
      FileStatus[] statuses = listStatus(f);

      if (statuses.length == 0) {
        // has no children
        String key = Path.getPathWithoutSchemeAndAuthority(f).toString();
        if (key.startsWith("/")) {
          key = key.substring(1);
        }
        if (!key.endsWith("/")) {
          key = key + "/";
        }
        logger.info("delete directory :" + key);
        deleteKey(key);
      } else {
        if (!recursive) {
          throw new IOException("dir not empty:" + f);
        } else {
          for (FileStatus st : statuses) {
            delete(st.getPath(), recursive);
          }
          delete(f, false);
        }
      }
    }
    return true;
  }

  private void deleteKey(String key) throws IOException {
    ObjectApiBuilder apiBuilder = UfileClient.object(objectAuthorization, config);
    try {
      apiBuilder.deleteObject(key, bucket).execute();
    } catch (Exception ex) {
      throw new IOException(ex);
    }
  }


  public boolean isFile(Path f) throws IOException {
    FileStatus fileStatus = getFileStatus(f);
    return fileStatus.isFile();
  }

  /**
   * list status.
   *
   * @param f file path
   * @return file statuses
   * @throws FileNotFoundException ex
   * @throws IOException ioex
   */
  @Override
  public FileStatus[] listStatus(Path f) throws FileNotFoundException, IOException {

    String path = Path.getPathWithoutSchemeAndAuthority(f).toString();

    if (path.startsWith("/")) {
      path = path.substring(1);
    }
    path = path.endsWith("/") ? path : path + "/";
    String prefix = path.equals("/") ? "" : path;
    ObjectApiBuilder apiBuilder = UfileClient.object(objectAuthorization, config);
    String nextMarker = "";
    List<FileStatus> statusList = new LinkedList<>();
    do {
      try {
        ObjectListWithDirFormatApi listApi =
              apiBuilder.objectListWithDirFormat(bucket)
                    .withPrefix(prefix).withMarker(nextMarker)
                    .withDelimiter("/").dataLimit(200);
        ObjectListWithDirFormatBean bean = listApi.execute();
        List<CommonPrefix> commonPrefixes = bean.getCommonPrefixes();
        if (commonPrefixes != null && commonPrefixes.size() > 0) {
          //dirs
          for (CommonPrefix cp : commonPrefixes) {
            if (cp.getPrefix().equals(path) || cp.getPrefix().endsWith("//")) {
              logger.warn("bad object key detected:" + cp.getPrefix());
              continue;
            }
            FileStatus status = new FileStatus(0, true,
                  3,
                  128, 0L,
                  0L, FsPermission.getDefault(), "",
                  "", new Path("/" + cp.getPrefix()));
            statusList.add(status);
          }
        }
        List<ObjectContentBean> contentBeanList = bean.getObjectContents();
        //files
        for (ObjectContentBean cb : contentBeanList) {
          //ignore dir object,eg: a/b/c/
          if (cb.getKey().endsWith("/")) {
            continue;
          }
          long length = Long.parseLong(cb.getSize());
          FileStatus status = new FileStatus(length, false,
                3,
                128, cb.getLastModified() * 1000,
                0L, FsPermission.getDefault(), "",
                "", new Path("/" + cb.getKey()));
          statusList.add(status);
        }
        nextMarker = bean.getNextMarker();

      } catch (Exception e) {
        throw new IOException(e);
      }
    } while (!Strings.isNullOrEmpty(nextMarker));

    //if (statusList.size() == 0) {
    //  try {
    //    logger.info("path.substring(1) ==========>  " + path.substring(1));
    //    apiBuilder.objectProfile(path.substring(1), bucket).execute();
    //  } catch (Exception e) {
    //    throw new FileNotFoundException(f.toString());
    //  }
    //}
    FileStatus[] statuses = statusList.toArray(new FileStatus[statusList.size()]);
    return statuses;
  }

  @Override
  public Path getWorkingDirectory() {
    return workingDir;
  }

  @Override
  public void setWorkingDirectory(Path newDir) {
    this.workingDir = newDir;
  }

  private void mkdir(Path f) throws IOException {
    try {
      FileStatus status = getFileStatus(f);
      if (status.isFile()) {
        throw new IOException("file exists");
      }
      return;
    } catch (FileNotFoundException e) {
      //
    }
    String path = Path.getPathWithoutSchemeAndAuthority(f).toString();
    if (path.startsWith("/")) {
      path = path.substring(1);
    }
    if (!path.endsWith("/")) {
      path = path + "/";
    }
    try {
      ObjectApiBuilder apiBuilder = UfileClient.object(objectAuthorization, config);
      InputStream in = new InputStream() {
        @Override
        public int read() throws IOException {
          return -1;
        }
      };
      apiBuilder.putObject(in, 0L,
            "application/octet-stream").toBucket(bucket).nameAs(path).execute();
    } catch (Exception ex) {
      throw new IOException(ex);
    }
  }

  @Override
  public boolean mkdirs(Path f, FsPermission permission) throws IOException {
    ObjectApiBuilder apiBuilder = UfileClient.object(objectAuthorization, config);
    FileStatus status = null;
    try {
      status = getFileStatus(f);
      throw new IOException("file or directory exists");
    } catch (FileNotFoundException nof) {
      boolean parentExist = exists(f.getParent());
      if (!parentExist) {
        mkdirs(f.getParent());
      } else if (getFileStatus(f.getParent()).isFile()) {
        throw new IOException("path is file:" + f.getParent().toString());
      }
      mkdir(f);
    }
    return true;
  }

  /**
   * get file status.
   *
   * @param f file path
   * @return file status
   * @throws IOException ex
   */
  @Override
  public FileStatus getFileStatus(Path f) throws IOException {
    String path = Path.getPathWithoutSchemeAndAuthority(f).toString();
    if (path.equals("/")) {
      return new FileStatus(0L,
            true,
            0,
            128, 0,
            0L, FsPermission.getDefault(), "",
            "", new Path("/"));
    }
    path = pathToKey(f);

    ObjectApiBuilder apiBuilder = UfileClient.object(objectAuthorization, config);
    ObjectListApi listApi = apiBuilder.objectList(bucket).withPrefix(path).withMarker(null);
    ObjectListBean listBean = null;
    try {
      listBean = listApi.execute();
    } catch (Exception e) {
      throw new IOException(e);
    }
    if (listBean.getObjectList().size() == 0) {
      throw new FileNotFoundException(f.toString());
    }
    FileStatus fileStatus = null;
    while (true) {
      for (ObjectInfoBean bean : listBean.getObjectList()) {
        String fileName = bean.getFileName();
        if (fileName.endsWith("/")) {
          fileName = fileName.substring(0, fileName.length() - 1);
        }
        if (fileName.equals(path)) {
          fileStatus = infoBeanToStatus(bean);
          break;
        } else if (ArrayUtil.startWith(path.split("/"), fileName.split("/"))) {
          fileStatus = new FileStatus(0L,
                true,
                0,
                128, bean.getModifyTime() * 1000,
                0L, FsPermission.getDefault(), "",
                "", new Path(path));
          break;
        }
      }
      if (fileStatus == null && listBean.getNextMarker().length() > 0) {
        try {
          listBean = apiBuilder.objectList(bucket).withPrefix(path)
                .withMarker(listBean.getNextMarker())
                .execute();
        } catch (Exception e) {
          throw new IOException(e);
        }
      } else {
        break;
      }
    }
    if (fileStatus != null) {
      return fileStatus;
    } else {
      throw new FileNotFoundException(f.toString());
    }
  }

  private FileStatus infoBeanToStatus(ObjectInfoBean bean) {
    String path = "/" + bean.getFileName();
    FileStatus status = new FileStatus(bean.getSize(),
          bean.getFileName().endsWith("/"),
          3,
          128, bean.getModifyTime() * 1000,
          0L, FsPermission.getDefault(), "",
          "", new Path(path));
    return status;
  }

  private String pathToKey(Path p) {
    String key = Path.getPathWithoutSchemeAndAuthority(p).toString();
    if (key.startsWith("/")) {
      key = key.substring(1);
    }
    if (key.endsWith("/")) {
      key = key.substring(0, key.length() - 1);
    }
    return key;
  }

}
